package com.flute.haflute.agent.heartbeat;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Random;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.mina.core.session.IoSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.flute.haflute.agent.topo.ClusterMessageListener;
import com.flute.haflute.agent.topo.ClusterTopoDBService;
import com.flute.haflute.net.messaging.Message;
import com.flute.haflute.net.messaging.MessageSessionPair;
import com.flute.haflute.net.messaging.UdpComparableSession;
import com.flute.haflute.tools.Utils;

/**
 * DM、SM和BDM使用的心跳监听器，监听所有成员的心跳消息<br>
 * 心跳监听的处理有多个后台线程:<br>
 * 
 * @see HeartBeatMessageSinkThread<br>
 *      对于每个session的请求，一定由同一个处理线程处理
 */
public class HeartBeatListener4Master implements ClusterMessageListener {

	private static Logger logger = LoggerFactory
			.getLogger(HeartBeatListener4Master.class);

	/** maxSinkers 处理心跳消息的最多后台线程数目 */
	private static int maxSinkers = 5;

	/** cachedSession2Sinkers <Session, 心跳处理线程>对的缓存，保证统一session交给同一个心跳处理线程处理 */
	private ConcurrentHashMap<UdpComparableSession, HeartBeatMessageSinkThread> cachedSession2Sinkers;

	/** allSinkers 心跳消息处理线程 */
	private HeartBeatMessageSinkThread[] allSinkers;

	/** cleanThread 清理session池 */
	private SessionCleanThread cleanThread;

	private Random rand;

	public HeartBeatListener4Master() {
		cachedSession2Sinkers = new ConcurrentHashMap<UdpComparableSession, HeartBeatMessageSinkThread>(
				2 * maxSinkers);
		cleanThread = new SessionCleanThread("Session Clean Thread");

		rand = new Random(50);
		allSinkers = new HeartBeatMessageSinkThread[maxSinkers];
		for (int i = 0; i < maxSinkers; i++) {
			allSinkers[i] = new HeartBeatMessageSinkThread(
					"HeartBeat Sinker No." + i);
		}
		startDaemons();
	}

	private void startDaemons() {
		cleanThread.start();
		for (int i = 0; i < maxSinkers; i++) {
			allSinkers[i].start();
		}
	}

	/*
	 * @see com.flute.haflute.clusterservice.StateChangeListener#receiveUpdate(
	 * com.flute.haflute.net.messaging.Message)
	 */
	@Override
	public void asyncHandle(Message message, IoSession session) {
		UdpComparableSession cSession = new UdpComparableSession(session);
		HeartBeatMessageSinkThread sinker = cachedSession2Sinkers.get(cSession);
		if (sinker == null) {
			sinker = allSinkers[rand.nextInt(100) % maxSinkers];
			cachedSession2Sinkers.put(cSession, sinker);
		}
		sinker.asyncHandle(new MessageSessionPair(message, session));
	}

	/**
	 * 处理线程，具体处理某个心跳消息
	 * 
	 */
	static class HeartBeatMessageSinkThread extends Thread {
		private LinkedBlockingQueue<MessageSessionPair> heartBeatCache;

		public HeartBeatMessageSinkThread(String name) {
			super(name);
			heartBeatCache = new LinkedBlockingQueue<MessageSessionPair>();
		}

		public void asyncHandle(MessageSessionPair oneHeartBeat) {
			heartBeatCache.offer(oneHeartBeat);
		}

		@Override
		public void run() {
			while (true) {
				doHandle(heartBeatCache.poll());
				try {
					sleep(100);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}

		private void doHandle(MessageSessionPair nodeHeartBeat) {
			if (nodeHeartBeat == null)
				return;

			Message message = nodeHeartBeat.getMessage();
			IoSession session = nodeHeartBeat.getSession();

			NodeStateUpdateResult result = ClusterTopoDBService.getInstance()
					.updateNodeState(message);
			if (!result.isFinished()) {
				try {
					Utils.sendMessage(session, result.getMessage());
				} catch (IOException e) {
					logger.error(getName(), e);
				}
			}
		}
	}

	/**
	 * 清理失效session的线程
	 * 
	 */
	class SessionCleanThread extends Thread {
		public SessionCleanThread(String name) {
			super(name);
		}

		@Override
		public void run() {
			while (true) {
				clean();
			}
		}

		public void clean() {
			ArrayList<UdpComparableSession> disconnectedSessions = new ArrayList<UdpComparableSession>(
					cachedSession2Sinkers.size() / 60 + 1);
			for (Entry<UdpComparableSession, HeartBeatMessageSinkThread> entrySet : cachedSession2Sinkers
					.entrySet()) {
				UdpComparableSession session = entrySet.getKey();
				if (!session.getSession().isConnected()) {
					disconnectedSessions.add(session);
				}
			}

			for (UdpComparableSession session : disconnectedSessions) {
				cachedSession2Sinkers.remove(session);
			}

			waiting();
		}

		private void waiting() {
			try {
				sleep(1000L);
			} catch (InterruptedException e) {
			}

		}

	}

}
